package com.zhaolaobao.redis.queue.core;

import com.zhaolaobao.redis.queue.config.MethodQuote;
import com.zhaolaobao.redis.service.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.lang.reflect.InvocationTargetException;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName RedisDelayQueueListen
 * @Description //
 * @Author WHX
 * @Date 2021/10/29 14:46
 **/
@Component
@Slf4j
public class RedisDelayQueueListen {

    @Resource
    private RedisService redisService;
    @Resource
    private RedisQueueTemplate redisQueueTemplate;

    public void listen(String queueName, MethodQuote methodQuote) {
        while (true) {
            try {
                doListener(queueName, methodQuote);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private void doListener(String queueName, MethodQuote methodQuote) throws InvocationTargetException, IllegalAccessException, InterruptedException {
        String threadName = Thread.currentThread().getName();
        // 获取元素：已到期的延时任务集合
        Set<Object> msgs = redisService.zsRangeByScore(queueName, 0, System.currentTimeMillis());
        log.debug(threadName + "获取到期消息：" + msgs.size() + "条");
        if (msgs.isEmpty()) {
            TimeUnit.SECONDS.sleep(1);
            return;
        }
        Iterator<Object> it = msgs.iterator();
        while (it.hasNext()) {
            Object obj = it.next();
            // 删除redis元素成功的服务，表示抢到了任务
            if (redisService.zsRemove(queueName, obj) > 0) {
                log.debug(threadName + "当前服务抢到消息，成功删除redis中的本条消息" + obj.toString());
                // 同步处理业务逻辑，同一队列的消息按zset排序依次消费：
                try {
                    methodQuote.getMethod().invoke(methodQuote.getObject(), obj);
                } catch (Exception e) {
                    redisQueueTemplate.sendDelayQueue(queueName, obj, 1);
                    log.error("redis延时队列监听方法调用失败，请检查监听方法，当前消息已重入队列，scope为1； queueName:{}, msg:{}" + queueName, obj);
                    throw e;
                }
            } else {
                log.debug(threadName + "消息已被删除，当前服务未抢到消息");
            }
        }
    }

}
